package com.dxf.bigdata.D04_spark

import org.apache.commons.logging.{Log, LogFactory}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *  RDD方法主要有 转换 和 行动
 *   转换 如  Map , flatMap
 *   行动 collect
 *
 *
 *   RDD算子:   RDD的方法就是算子 --> 对数据的操作称为算子
 *
 *   RDD算子 => 转换算子 Map , 行动算子  collect
 *
 *   ===========================
 *
 *
 *   Map  => 处理数据,一个处理完了处理下一个 ==> 效率肯定底 (对比字节流) ==> 处理方式 缓存区
 *
 *
 *
 *
 *
 */
object T10_RDD_Map {
  private val log: Log = LogFactory.getLog(T08_RDD使用_从日志中读取uri.getClass)
  def main(args: Array[String]): Unit = {

    //TODO 准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    //TODO 创建RDD
    //查看单分区和多分区的执行结果,
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)


    /**
     * >>>>
     * >>>>
     *  为什么?
     *   mapPartitions
     *   1 内部封装迭代器,提高效率
     *   2 会把每个分区中的所有的数据都取出来再操作,
     *   3 rdd中有2个分区, mapPartitions会对每个分区生成迭代器 ,然后进行操作
     *   4 缺点   iter.map(_ * 2)  是对迭代器中的所有数据进行操作,
     *     把整个分区中的数据加载进内存,但是处理完的数据,存在对象的引用不会别释放,内存消耗大
     *
     *
     */
    val rddmap: RDD[Int] = rdd.mapPartitions(iter => {
      println(">>>>")
      iter.map(_ * 2)
    })
    rddmap.collect()


    //取出每个分区的最大值
    //rdd 2 分区 [1,2] [3,4]
    val rddmap2: RDD[Int] = rdd.mapPartitions(iter => {
     List(iter.max).iterator
    })
    rddmap2.collect().foreach(println(_))


    //TODO 关闭环境
    sc.stop()


  }
}
